// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#include <pollux/dwio/dwrf/common/file_metadata.h>

namespace kumo::pollux::dwrf {
namespace detail {
using common::CompressionKind;

CompressionKind orcCompressionToCompressionKind(
    proto::orc::CompressionKind compression) {
  switch (compression) {
    case proto::orc::CompressionKind::NONE:
      return CompressionKind::CompressionKind_NONE;
    case proto::orc::CompressionKind::ZLIB:
      return CompressionKind::CompressionKind_ZLIB;
    case proto::orc::CompressionKind::SNAPPY:
      return CompressionKind::CompressionKind_SNAPPY;
    case proto::orc::CompressionKind::LZO:
      return CompressionKind::CompressionKind_LZO;
    case proto::orc::CompressionKind::LZ4:
      return CompressionKind::CompressionKind_LZ4;
    case proto::orc::CompressionKind::ZSTD:
      return CompressionKind::CompressionKind_ZSTD;
  }
  POLLUX_FAIL("Unknown compression kind: {}", CompressionKind_Name(compression));
}
} // namespace detail

TypeKind TypeWrapper::kind() const {
  if (format_ == DwrfFormat::kDwrf) {
    switch (dwrfPtr()->kind()) {
      case proto::Type_Kind_BOOLEAN:
      case proto::Type_Kind_BYTE:
      case proto::Type_Kind_SHORT:
      case proto::Type_Kind_INT:
      case proto::Type_Kind_LONG:
      case proto::Type_Kind_FLOAT:
      case proto::Type_Kind_DOUBLE:
      case proto::Type_Kind_STRING:
      case proto::Type_Kind_BINARY:
      case proto::Type_Kind_TIMESTAMP:
        return static_cast<TypeKind>(dwrfPtr()->kind());
      case proto::Type_Kind_LIST:
        return TypeKind::ARRAY;
      case proto::Type_Kind_MAP:
        return TypeKind::MAP;
      case proto::Type_Kind_UNION: {
        DWIO_RAISE("Union type is deprecated!");
      }
      case proto::Type_Kind_STRUCT:
        return TypeKind::ROW;
      default:
        POLLUX_FAIL("Unknown type kind: {}", Type_Kind_Name(dwrfPtr()->kind()));
    }
  }

  switch (orcPtr()->kind()) {
    case proto::orc::Type_Kind_BOOLEAN:
    case proto::orc::Type_Kind_BYTE:
    case proto::orc::Type_Kind_SHORT:
    case proto::orc::Type_Kind_INT:
    case proto::orc::Type_Kind_LONG:
    case proto::orc::Type_Kind_FLOAT:
    case proto::orc::Type_Kind_DOUBLE:
    case proto::orc::Type_Kind_STRING:
    case proto::orc::Type_Kind_BINARY:
    case proto::orc::Type_Kind_TIMESTAMP:
      return static_cast<TypeKind>(orcPtr()->kind());
    case proto::orc::Type_Kind_LIST:
      return TypeKind::ARRAY;
    case proto::orc::Type_Kind_MAP:
      return TypeKind::MAP;
    case proto::orc::Type_Kind_UNION: {
      DWIO_RAISE("Union type is deprecated!");
    }
    case proto::orc::Type_Kind_STRUCT:
      return TypeKind::ROW;
    case proto::orc::Type_Kind_VARCHAR:
      return TypeKind::VARCHAR;
    // Date is a logical type of INTEGER (for the number of days since EPOCH).
    case proto::orc::Type_Kind_DATE:
      return TypeKind::INTEGER;
    case proto::orc::Type_Kind_DECIMAL: {
      if (orcPtr()->precision() <= pollux::ShortDecimalType::kMaxPrecision) {
        return TypeKind::BIGINT;
      } else {
        return TypeKind::HUGEINT;
      }
    }
    case proto::orc::Type_Kind_CHAR:
    case proto::orc::Type_Kind_TIMESTAMP_INSTANT:
      POLLUX_FAIL(fmt::format(
          "{} not supported yet.",
          proto::orc::Type_Kind_Name(orcPtr()->kind())));
    default:
      POLLUX_FAIL("Unknown type kind: {}", Type_Kind_Name(orcPtr()->kind()));
  }
}

common::CompressionKind PostScript::compression() const {
  return format_ == DwrfFormat::kDwrf
      ? static_cast<common::CompressionKind>(dwrfPtr()->compression())
      : detail::orcCompressionToCompressionKind(orcPtr()->compression());
}

} // namespace kumo::pollux::dwrf
